package org.fujene.search;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import org.fujene.communicate.blockwise.BlockwisePeerSystem;
import org.fujene.structure.Constants;
import org.fujene.structure.ExtendQueue;
import org.fujene.structure.ExtendResult;
import org.fujene.structure.SearchQueue;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.NilType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Request;
import org.msgpack.rpc.Server;
import org.msgpack.rpc.loop.EventLoop;

public class BlockwiseSearchSystem {
	// executor専用
	public int EXECUTOR_THREADS = 1;
	public boolean stopFlag = false;
	public volatile int threadCounter = 0;
	private BlockwisePeerSystem myPeer = null;
	private ExecutorService exec = null;
	private Logger logger = null;

	public BlockwiseSearchSystem(BlockwisePeerSystem myPeer, Logger logger) {
		assert logger != null && myPeer != null;

		// 待ちうけスレッド起動
		exec = Executors.newSingleThreadExecutor();
		exec.execute(new SearchMainService());
		this.myPeer = myPeer;
		this.logger = logger;
	}

	public class SearchRPCHandler {
		public void search(Request req) throws Exception {
			MessagePackObject obj = req.getArguments();
			MessagePackObject[] args = obj.asArray();
			if (args.length < 4
					&& (!args[0].isRawType() || !args[1].isRawType()
							|| !args[2].isIntegerType() || !args[3]
							.isIntegerType())) {
				logger.severe("Please post appropriate query.");
			}

			try {
				// Future threadで投げたクエリと戻りの対応をとる
				ExtendResult result;
				result = myPeer.setNewSearch(new SearchQueue(args[0]
						.asString(), args[1].asString(), args[2].asInt(),
						args[3].asInt()));

				MessagePackObject[] objs = new MessagePackObject[result.getLength()];
				int count = 0;
				int i;
				for (ExtendQueue eachQueue : result.getQueues()) {
					for (i = 0; i < eachQueue.contents.length; i++) {
						objs[count++] = RawType.create(eachQueue.contents[i]);
					}
				}

				req.sendResult(ArrayType.create(objs));
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		public void stop(Request req) {
			this.notifyAll();
			exec.shutdownNow();
			req.sendResult(NilType.create());
		}
	}

	public class SearchMainService implements Runnable {
		public SearchMainService() {
		}

		@Override
		public void run() {
			logger.fine("SearchMainService started.");
			Server server = new Server(EventLoop.defaultEventLoop());
			try {
				server.serve(new SearchRPCHandler());
				server.listen(Constants.SEARCH_PORT);

				logger.fine("Search server port " + Constants.SEARCH_PORT
						+ " has established.");

				synchronized (this) {
					this.wait();
				}

				logger.fine("SearchMainService ended.");

				server.close();
			} catch (Exception ignored) {
			}
		}
	}
}
